package com.fwmagic.flink.tablesql.batch;

import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.java.BatchTableEnvironment;

public class BatchSqlWordCount {
    public static void main(String[] args) throws Exception{
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        BatchTableEnvironment batchTableEnv = BatchTableEnvironment.create(env);

        DataSet<BatchTableWordCount.WordCount> dataSet = env.fromElements(
                new BatchTableWordCount.WordCount("Spark", 1L),
                new BatchTableWordCount.WordCount("Spark", 1L),
                new BatchTableWordCount.WordCount("Flink", 1L),
                new BatchTableWordCount.WordCount("Flink", 1L),
                new BatchTableWordCount.WordCount("Flink", 1L),
                new BatchTableWordCount.WordCount("Flink", 1L),
                new BatchTableWordCount.WordCount("Java", 1L),
                new BatchTableWordCount.WordCount("Spark", 1L)
        );

        batchTableEnv.registerDataSet("t_wc", dataSet, "word, counts");

        Table table = batchTableEnv.sqlQuery("select word,sum(counts) as counts from t_wc group by word having sum(counts) >2 order by counts asc");

        DataSet<BatchTableWordCount.WordCount> wordCountDataSet = batchTableEnv.toDataSet(table, BatchTableWordCount.WordCount.class);

        wordCountDataSet.print();

    }



    public static class WordCount {
        public String word;

        public Long counts;

        public WordCount() {
        }

        public WordCount(String word, Long counts) {
            this.word = word;
            this.counts = counts;
        }

        @Override
        public String toString() {
            return "WordCount{" +
                    "word='" + word + '\'' +
                    ", counts=" + counts +
                    '}';
        }
    }
}
